package com.sunzm.flink.datastream.scala.source.kafka.deserializer

import java.nio.charset.StandardCharsets

import com.alibaba.fastjson.JSONObject
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema
import org.apache.flink.streaming.api.scala.createTypeInformation
import org.apache.flink.util.Collector
import org.apache.kafka.clients.consumer.ConsumerRecord

/**
 *
 * ${DESCRIPTION}
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-12-24 0:03
 */
class MyKafkaRecordDeserializationSchema extends KafkaRecordDeserializationSchema[JSONObject]{
  override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]], out: Collector[JSONObject]): Unit = {


    val nObject = new JSONObject()

    nObject.put("topic", record.topic())
    nObject.put("value", new String(record.value(), StandardCharsets.UTF_8))

    out.collect(nObject)
  }

  override def getProducedType: TypeInformation[JSONObject] = {

    createTypeInformation[JSONObject]

  }
}
